#include "server.h"

#include <iostream>
#include <sstream>
#include <functional>

namespace crawler { namespace server {
CrawlerServerHandler::CrawlerServerHandler(Store *store, const Config *config) :
  stop_(false),
  store_(store),
  config_(config),
  homepage_(config->homepage()) {
  // insert homepage to the storage
  Link homelink;
  homelink.__set_url(homepage_);
  store_->StoreLink(homelink); 
}
void CrawlerServerHandler::GetDownloadUrl(std::string& _return) {
  CheckStop();
  boost::mutex::scoped_lock lock(mutex_);
  if (url_pool_.empty()) {
    std::vector<std::pair<int, std::string> > depth_queue;
    for (Store::PageIterator it=store_->Begin(); it!=store_->End(); ++it) {
      if (it->status != PageStatus::kNotYetDownload) continue;
      Url u(it->url);
      depth_queue.push_back(std::make_pair(u.GetDepth(), it->url));
      std::push_heap(depth_queue.begin(), depth_queue.end(),
                     std::greater<std::pair<int, std::string> >());
      if (depth_queue.size() > kPoolSize) depth_queue.pop_back();
    }
    for (std::vector<std::pair<int, std::string> >::iterator it=depth_queue.begin();
         it!=depth_queue.end(); ++it) {
      url_pool_.push_back(it->second);
    }
    std::cout << "url pool is empty, load " << url_pool_.size() << " urls from disk" << std::endl;
  }

  if (url_pool_.empty()) {
    ServerError error;
    error.__set_why("no more links in server!");
    throw error;
  } else {
    _return = url_pool_.front();
    url_pool_.pop_front();
    std::cout << "sending url to client, " << _return << std::endl;
  }
}
void CrawlerServerHandler::GetStoredPage(Page& _return, const std::string& url) {
  CheckStop();
  boost::mutex::scoped_lock lock(mutex_);
  try {
    store_->GetPage(url, _return);
  } catch (const std::exception &ex) {
    std::cout << "adding url to download list, " << url << std::endl;
    url_pool_.push_front(url);
    throw;
  }
  std::cout << "sending page to client, " << _return.url << std::endl;
}
void CrawlerServerHandler::SendPage(const Page& page) {
  CheckStop();
  std::cout << "received page from client, " << page.url << std::endl;
  if (page.content.find("Skip this ad") != std::string::npos &&
      page.content.find("requested will load in") != std::string::npos) return;
  store_->StorePage(page);
  if (page.status == PageStatus::kSuccessDownload) {
    PageParser parser(page);
    std::vector<Link> &links = parser.GetLinks();
    for (std::vector<Link>::iterator it=links.begin(); it!=links.end(); ++it) {
      if (it->url.find(homepage_) != 0) continue;
      if (it->url.find("javascript") != std::string::npos) continue;
      if (it->url.find("mailto:") != std::string::npos) continue;
      store_->StoreLink(*it);
    }
  }
}
void CrawlerServerHandler::GetProgress(Progress& _return) {
  CheckStop();
  std::cout << "sending progress to client" << std::endl;
  for (Store::PageIterator it=store_->Begin(); it!=store_->End(); ++it) {
    std::ostringstream oss;

    ++_return.total_urls;
    if (it->status == PageStatus::kSuccessDownload) {
      ++_return.downloaded_urls;
    }
    oss << "status:" << it->status << " " << it->url;
    _return.url_list.push_back(oss.str());
  }
}
void CrawlerServerHandler::Stop() {
  stop_ = true;
}
void CrawlerServerHandler::CheckStop() {
  if (stop_) {
    ServerError error;
    error.__set_why("Server is stopped, retry later");
    throw error;
  }
}
Server::Server(const Config *config) :
  config_(config),
  store_(new Store(config_->store_db_path())) {
  handler_.reset(new CrawlerServerHandler(store_.get(), config_));
  shared_ptr<TProcessor> processor(new CrawlerServerProcessor(handler_));
  shared_ptr<TServerTransport> serverTransport(new TServerSocket(config_->server_port()));
  shared_ptr<TTransportFactory> transportFactory(new TBufferedTransportFactory());
  shared_ptr<TProtocolFactory> protocolFactory(new TBinaryProtocolFactory());
  server_.reset(new TThreadedServer (processor, serverTransport,
                                     transportFactory, protocolFactory));
}
void Server::operator()() {
  server_->serve();
}
void Server::Start() {
  std::cout << "server is starting" << std::endl;
  thread_.reset(new boost::thread(*this));
}
void Server::Stop() {
  std::cout << "server is stopping" << std::endl;
  handler_->Stop();
  server_->stop();
}
void Server::Join() {
  thread_->join();
}
}}

